package com.dgw.springbootkafka.controller;

import com.dgw.springbootkafka.config.KafkaConfig;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @program: springboot-kafka
 * @description:
 * @author: Mr.Dai
 * @create: 2020-02-23 15:19
 **/
@RestController
public class KafkaTopicManager {

    @Autowired
    private KafkaConfig kafkaConfig;

    @GetMapping("createTopic")
    public void createTopic() {
        AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps());

        NewTopic newTopic = new NewTopic("test1", 4, (short) 1);
        Collection<NewTopic> newTopicList = new ArrayList<>();
        newTopicList.add(newTopic);
        adminClient.createTopics(newTopicList);

        adminClient.close();
    }

    @GetMapping("deleteTopic")
    public void deleteTopic() {
        AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps());
        adminClient.deleteTopics(Arrays.asList("test1"));
        adminClient.close();
    }

    @GetMapping("listAllTopic")
    public void listAllTopic() {
        AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps());
        ListTopicsResult result = adminClient.listTopics();
        KafkaFuture<Set<String>> names = result.names();
        try {
            names.get().forEach((k) -> {
                System.out.println(k);
            });
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        adminClient.close();
    }

    @GetMapping("getTopic")
    public void getTopic() {
        AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps());

        DescribeTopicsResult describeTopics = adminClient.describeTopics(Arrays.asList("test-syn"));

        try {
            describeTopics.all().get().forEach((k, v) -> System.out.println("k: " + k + " ,v: " + v.toString() + "\n"));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        adminClient.close();
    }
}
